package com.ibm.cps.demo;

import java.util.Collection;
import java.util.HashSet;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Preconditions;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.RawDataMessage;
import com.ibm.cps.messagequeue.AMQPClient;
import com.ibm.interfaces.IMessageClient;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;

/**
 * Subscribe hit messages
 * 
 * @author Jingjing Wang
 */
public class SubscribeMessages {
	private String tenantid;
	private IMessageClient consumer;
	private Collection<String> rtnTopicNames;

	protected Logger logger;
	private MongoPersistency persistency;

	public SubscribeMessages(String tenantid) throws CPSException {
		PropertyConfigurator.configure(LocalConfig.getLoggerProperties());
		logger = Logger.getLogger(SubscribeMessages.class);
		this.tenantid = tenantid;
		rtnTopicNames = getReturnTopics();

		try {
			Preconditions.checkNotNull(rtnTopicNames);
			Preconditions.checkArgument(rtnTopicNames.size() >= 1, "Null topic name to listen to.");
		} catch (Exception e) {
			throw new CPSException(ErrorCode.NULL, e.getMessage());
		}
		for (String topic : rtnTopicNames) {
			consumer.subscribe(topic);
		}
	}

	// private String getProtocolType() throws CPSException {
	// Collection<String> publishers = persistency.getMetadataByUserType(
	// tenantid, MessageFields.PUBLISH_TYPE);
	// if (publishers == null || publishers.size() < 1) {
	// return null;
	// }
	// String publisher = publishers.iterator().next();
	// ObjectNode node = JSONUtility.fromObjectString(publisher);
	// if (node.get(MessageFields.PROTOCOL_TYPE) == null) {
	// throw new CPSException(ErrorCode.INVALID_REQUEST_CODE,
	// "Invalid publisher stored in persistency.");
	// } else {
	// return node.get(MessageFields.PROTOCOL_TYPE).asText();
	// }
	// }

	private Collection<String> getReturnTopics() throws CPSException {
		persistency = MongoPersistency.getInstace();
		Collection<String> outputs = persistency.getProcessors(tenantid, null, MessageFields.PUBLISH_TYPE, null);
		Collection<String> topicNames = new HashSet<String>();
		if (outputs == null || outputs.size() < 1) {
			return null;
		}
		for (String output : outputs) {
			ObjectNode node = JSONUtility.fromObjectString(output);
			System.out.println(output);
			consumer = new AMQPClient(node.get(MessageFields.HOSTNAME).asText(), node.get(MessageFields.PORT).asInt(),
					node.get(MessageFields.USERNAME).asText(), node.get(MessageFields.PASSWORD).asText());
			if (node.get(MessageFields.TOPICNAME) == null) {
				throw new CPSException(ErrorCode.INVALID_REQUEST_CODE, "Invalid rule stored in persistency.");
			}
			topicNames.add(node.get(MessageFields.TOPICNAME).asText());
		}
		return topicNames;
	}

	public void subscribeMessage(Comparable tsKey) throws CPSException {
		Collection<String> mqMessages;
		while (true) {
			mqMessages = consumer.consumeMessage();
			if (mqMessages != null) {
				for (String mqMessage : mqMessages) {
					RawDataMessage rawDataMessage = new RawDataMessage(null, null, null, tsKey, mqMessage.getBytes());
					if (tsKey == rawDataMessage.getTsKey()) {
						logger.info(mqMessage);
					}
				}
			}
		}
	}

	public static void main(String[] args) {
		if (args == null || args.length != 2) {
			System.out.println("Usage: Program <tenantid tsKey>");
			return;
		}
		String tenantid = args[0];
		long tsKey = Long.parseLong(args[1]);
		SubscribeMessages subscriber;
		try {
			subscriber = new SubscribeMessages(tenantid);
			subscriber.subscribeMessage(tsKey);
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}
}
